package mqtt_cleint

import (
	"crypto/tls"
	"crypto/x509"
	"fmt"
	MQTT "github.com/eclipse/paho.mqtt.golang"
	"io/ioutil"
	"time"
)

type MqttClient struct {
	opts               *MQTT.ClientOptions
	Cli                MQTT.Client
	ConnectHandler     MQTT.OnConnectHandler
	ConnectLostHandler MQTT.ConnectionLostHandler
}

// ConnectTls tls方式连接服务器
func (obj *MqttClient) ConnectTls(address, clientId, caFile, certFile, keyFile string, port int) error {
	obj.opts = MQTT.NewClientOptions()
	broker := fmt.Sprintf("ssl://%s:%d", address, port)
	obj.opts.AddBroker(broker)
	obj.opts.SetClientID(clientId)
	obj.opts.OnConnect = obj.ConnectHandler
	obj.opts.OnConnectionLost = obj.ConnectLostHandler
	obj.opts.SetCleanSession(true)
	obj.opts.SetAutoReconnect(true)
	obj.opts.ConnectTimeout = 30 * time.Second

	caPool := x509.NewCertPool()
	ca, err := ioutil.ReadFile(caFile)
	if err != nil {
		panic("root ca read error:" + err.Error())
	}

	if !caPool.AppendCertsFromPEM(ca) {
		panic("root ca load error")
	}

	cert, err := tls.LoadX509KeyPair(certFile, keyFile)
	if err != nil {
		panic("key load error")
	}

	tlsConfig := &tls.Config{
		RootCAs:            caPool,
		ClientAuth:         tls.NoClientCert,
		ClientCAs:          nil,
		InsecureSkipVerify: false,
		Certificates:       []tls.Certificate{cert},
		MaxVersion:         tls.VersionTLS12,
		MinVersion:         tls.VersionTLS10,
	}

	obj.opts = obj.opts.SetTLSConfig(tlsConfig)
	obj.Cli = MQTT.NewClient(obj.opts)

	if token := obj.Cli.Connect(); token.Wait() && token.Error() != nil {
		return token.Error()
	}

	return nil
}

func (obj *MqttClient) ConnectPw(clientId, addr string, port int, user, pw string, reconnect bool) error {
	obj.opts = MQTT.NewClientOptions()
	obj.opts.AddBroker(fmt.Sprintf("tcp://%s:%d", addr, port))
	obj.opts.SetClientID(clientId)
	obj.opts.SetUsername(user)
	obj.opts.SetPassword(pw)
	obj.opts.SetAutoReconnect(reconnect)
	obj.opts.OnConnect = obj.ConnectHandler
	obj.opts.OnConnectionLost = obj.ConnectLostHandler
	obj.opts.SetConnectTimeout(20 * time.Second)
	//obj.opts.SetDefaultPublishHandler(f)

	obj.Cli = MQTT.NewClient(obj.opts)
	if obj.Cli.IsConnected() {
		return nil
	}
	if token := obj.Cli.Connect(); token.Wait() && token.Error() != nil {
		return token.Error()
	}
	return nil
}

func (obj *MqttClient) Disconnect() {
	obj.Cli.Disconnect(100)
}

func (obj *MqttClient) Close() {
	obj.Cli.Disconnect(0)
}

// Subscribe 订阅
func (obj *MqttClient) Subscribe(topic string, messageHandler MQTT.MessageHandler) error {

	if token := obj.Cli.Subscribe(topic, 0, messageHandler); token.Wait() && token.Error() != nil {
		return token.Error()
	}
	return nil
}
func (obj *MqttClient) Unsubscribe(topic string) error {
	if token := obj.Cli.Unsubscribe(topic); token.Wait() && token.Error() != nil {
		return token.Error()
	}
	return nil
}

// Publish 发布
func (obj *MqttClient) Publish(topic, content string) error {
	if token := obj.Cli.Publish(topic, 0, false, content); token.Wait() && token.Error() != nil {
		return token.Error()
	}
	return nil
}
